#include "ul_timer.h"
#include "ul_log_adapter.h"
#include "ul_malloc.h"
#include "ul_util.h"
#include "ul_atomic.h"
#include <stdlib.h>
#include <gtest/gtest.h>
#include <iostream>
#include <vector>
#include "ul_pipe_task_queue.h"
#include "ul_condition_task_queue.h"
using namespace LinuxNetworkUtility;
using namespace testing;
using namespace std;
char g_time_buf[sizeof(UlTimer)];
UlTimer *g_time = NULL;

char g_memory_pool_buf[sizeof(UlMemoryPool)];
UlMemoryPool *g_memory_pool = NULL;
UlModSet  g_mod_set;

void  __attribute__((constructor)) init_global_memory_pool()
{
  g_time = new(g_time_buf)UlTimer();
  EXPECT_EQ(g_time->init(10000),0);
  g_memory_pool = new(g_memory_pool_buf)UlMemoryPool(*g_time);
}

void  __attribute__((destructor)) destroy_global_memory_pool()
{
  g_memory_pool->~UlMemoryPool();
  g_memory_pool  = NULL;
  g_time->~UlTimer();
  g_time = NULL;
}

struct Tsk
{
  int64_t size_;
  char    buf_[0];
};

int64_t  total_task_num = 0;
int64_t  finished_task_num = 0;
UlConditionQueenWorkerTQ tsk_queue;
bool run = true;

void *worker(void * arg)
{
  const int32_t tsk_num = 8192;
  int64_t got_tsk_num = 0;
  int64_t worker_id = (int64_t)arg;
  void *tsks[tsk_num];
  while (run)
  {
    got_tsk_num = tsk_queue.claim_task(tsks, tsk_num, 10000, worker_id);
    if (got_tsk_num < 0)
    {
      EXPECT_TRUE(errno == EINTR || errno == EAGAIN);
    }
    else
    {
      ///UL_LOG_DEBUG("got tsk num:%d", got_tsk_num);
      for (int32_t i = 0; i < got_tsk_num; i++)
      {
        tsk_queue.free_tsk(tsks[i]);
        tsks[i] = NULL;
        /// {
        ///   static int64_t max_tsk_size = (2<<20);
        ///   long randint = random();
        ///   long tsk_size = randint%max_tsk_size + 1;
        ///   Tsk *result = (Tsk*)tsk_queue.malloc_tsk(tsk_size + sizeof(Tsk),0);
        ///   EXPECT_NE(result, (void*)0);
        ///   if (NULL != result)
        ///   {
        ///     EXPECT_EQ(tsk_queue.submit_output(result, worker_id),0);
        ///   }
        /// }
      }
    }
  }
  return NULL;
}

void usage(char **argv)
{
  fprintf(stderr, "%s <#worker> <#total_task>\n", argv[0]);
  exit(0);
}

int main(int argc, char **argv)
{
  if (argc != 3)
  {
    usage(argv);
  }
  int32_t worker_num = atoi(argv[1]);
  total_task_num = atoi(argv[2]);
  if (worker_num <= 0 || total_task_num <= 0)
  {
    usage(argv);
  }

  EXPECT_EQ(ul_init_malloc(*g_memory_pool,*g_time,g_mod_set,10000),0);  
  pthread_t threads[worker_num];  

  EXPECT_EQ(tsk_queue.init(worker_num),0);

  /// create producer
  for (int64_t i = 0; i < worker_num; i++)
  {
    pthread_create(threads+i, NULL, worker, (void*)i);
  }
  /// producer
  int64_t begin = ul_gettimeofday();
  int64_t dispatched_tsk_num = 0;
  int64_t got_result_num = 0;
  //for (int32_t i = 0; i < total_task_num; i++)
  got_result_num = total_task_num;
  while (got_result_num < total_task_num || dispatched_tsk_num < total_task_num)
  {
    if (dispatched_tsk_num < total_task_num)
    {
      static int64_t max_tsk_size = (2<<20);
      long randint = random();
      long tsk_size = randint%max_tsk_size + 1;
      Tsk *task = (Tsk*)tsk_queue.malloc_tsk(tsk_size + sizeof(Tsk),0);
      EXPECT_NE(task, (void*)0);
      if (NULL != task)
      {
        EXPECT_EQ(tsk_queue.dispatch_task(task),0);
      }
      dispatched_tsk_num ++;
    }
    /// if ((dispatched_tsk_num % worker_num*4 == 0 || dispatched_tsk_num >= total_task_num)
    ///     && got_result_num < total_task_num)
    /// {
    ///   /// got results
    ///   const int32_t tsk_num = 8192;
    ///   int64_t got_tsk_num = 0;
    ///   void *tsks[tsk_num];
    ///   got_tsk_num = tsk_queue.fetch_output(tsks, tsk_num, 0);
    ///   if (got_tsk_num < 0)
    ///   {
    ///     EXPECT_TRUE(errno == EINTR || errno == EAGAIN);
    ///   }
    ///   else
    ///   {
    ///     ///UL_LOG_DEBUG("got tsk num:%d", got_tsk_num);
    ///     for (int32_t i = 0; i < got_tsk_num; i++)
    ///     {
    ///       tsk_queue.free_tsk(tsks[i]);
    ///       tsks[i] = NULL;
    ///     }
    ///     got_result_num += got_tsk_num;
    ///   }
    /// }
  } 
  run = false;

  int64_t used_time = ul_gettimeofday() - begin;
  fprintf(stderr, "used_time:%ld\n", used_time);

  /// join producer
  for (int32_t i = 0; i < worker_num; i++)
  {
    pthread_join(threads[i], NULL);
  }
}

